package org.example.marker.service.impl;

import org.example.marker.buffer.BufferPaddingExecutor;
import org.example.marker.buffer.RejectedPutBufferHandler;
import org.example.marker.buffer.RejectedTakeBufferHandler;
import org.example.marker.buffer.RingBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Wills
 * @date 2020/05/26
 */
public class CachedUidGeneratorServiceImpl extends DefaultUidGeneratorServiceImpl implements DisposableBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(CachedUidGeneratorServiceImpl.class);
    private static final int DEFAULT_BOOST_POWER = 3;

    private int boostPower = DEFAULT_BOOST_POWER;
    private int paddingFactor = RingBuffer.DEFAULT_PADDING_PERCENT;
    private Long scheduleInterval;

    private RejectedPutBufferHandler rejectedPutBufferHandler;
    private RejectedTakeBufferHandler rejectedTakeBufferHandler;

    private RingBuffer ringBuffer;
    private BufferPaddingExecutor bufferPaddingExecutor;

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();
        this.initRingBuffer();
        LOGGER.info("Initialized RingBuffer successfully.");
    }

    @Override
    public long getUID(long appId) {
        if (appId > this.bitsAllocator.getMaxAppId()) {
            throw new RuntimeException("appId cannot max than " + this.bitsAllocator.getMaxAppId());
        }
        return appId << this.bitsAllocator.getAppShift() | this.ringBuffer.take();
    }

    @Override
    public String parseUID(long uid) {
        return super.parseUID(uid);
    }

    @Override
    public void destroy() {
        this.bufferPaddingExecutor.shutdown();
    }

    protected List<Long> nextIdsForOneSecond(long currentSecond) {
        int listSize = (int) this.bitsAllocator.getMaxSequence() + 1;
        List<Long> uidList = new ArrayList<>(listSize);

        long firstSeqUid = this.bitsAllocator.allocateWithoutAppId(currentSecond - this.bitsAllocator.getEpochSeconds(), this.workerId, 0L);
        for (int offset = 0; offset < listSize; offset++) {
            uidList.add(firstSeqUid + offset);
        }

        return uidList;
    }

    private void initRingBuffer() {
        // initialize RingBuffer
        int bufferSize = ((int) this.bitsAllocator.getMaxSequence() + 1) << this.boostPower;
        this.ringBuffer = new RingBuffer(bufferSize, this.paddingFactor);
        LOGGER.info("Initialized ring buffer size:{}, paddingFactor:{}", bufferSize, this.paddingFactor);

        // initialize RingBufferPaddingExecutor
        boolean usingSchedule = (this.scheduleInterval != null);
        this.bufferPaddingExecutor = new BufferPaddingExecutor(this.ringBuffer, this::nextIdsForOneSecond, usingSchedule);
        if (usingSchedule) {
            this.bufferPaddingExecutor.setScheduleInterval(this.scheduleInterval);
        }

        LOGGER.info("Initialized BufferPaddingExecutor. Using schdule:{}, interval:{}", usingSchedule, this.scheduleInterval);

        // set rejected put/take handle policy
        this.ringBuffer.setBufferPaddingExecutor(this.bufferPaddingExecutor);
        if (this.rejectedPutBufferHandler != null) {
            this.ringBuffer.setRejectedPutHandler(this.rejectedPutBufferHandler);
        }
        if (this.rejectedTakeBufferHandler != null) {
            this.ringBuffer.setRejectedTakeHandler(this.rejectedTakeBufferHandler);
        }

        // fill in all slots of the RingBuffer
        this.bufferPaddingExecutor.paddingBuffer();

        // start buffer padding threads
        this.bufferPaddingExecutor.start();
    }

    /**
     * Setters for spring property
     */
    public void setBoostPower(int boostPower) {
        Assert.isTrue(boostPower > 0, "Boost power must be positive!");
        this.boostPower = boostPower;
    }

    public void setRejectedPutBufferHandler(RejectedPutBufferHandler rejectedPutBufferHandler) {
        Assert.notNull(rejectedPutBufferHandler, "RejectedPutBufferHandler can't be null!");
        this.rejectedPutBufferHandler = rejectedPutBufferHandler;
    }

    public void setRejectedTakeBufferHandler(RejectedTakeBufferHandler rejectedTakeBufferHandler) {
        Assert.notNull(rejectedTakeBufferHandler, "RejectedTakeBufferHandler can't be null!");
        this.rejectedTakeBufferHandler = rejectedTakeBufferHandler;
    }

    public void setScheduleInterval(long scheduleInterval) {
        Assert.isTrue(scheduleInterval > 0, "Schedule interval must positive!");
        this.scheduleInterval = scheduleInterval;
    }

}
